package com.simple.datax.seda.core;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.simple.datax.seda.api.ConfigInfoIF;
import com.simple.datax.seda.api.DealSourceIF;
import com.simple.datax.seda.api.EventHandlerIF;
import com.simple.datax.seda.api.QueueElementIF;
import com.simple.datax.seda.api.StageWrapperIF;
import com.simple.datax.seda.core.event.EntityNestedEvent;
import com.simple.datax.seda.core.event.HeartbeatEvent;
import com.simple.datax.seda.core.event.NullEvent;
import com.simple.datax.seda.core.event.StageInterruptEvent;
import com.simple.datax.seda.core.exception.InitFailedException;
import com.simple.datax.seda.core.exception.UnknownStageEventException;

public class DealSourceEventHandler implements EventHandlerIF {
	private Log LOG = LogFactory.getLog(DealSourceEventHandler.class);
	private DealSourceIF dealSource = null;
	private StageWrapperIF nextStage;
	private long sleepMilisecond = 1000;

	public DealSourceEventHandler() {

	}

	public void clear() {
		dealSource.clear();
	}

	public void initialize() throws InitFailedException {
		if (nextStage == null) {
			throw new InitFailedException("nextStage属性是空");
		}
		if (dealSource == null) {
			throw new InitFailedException("dealSource属性是空");
		}
		dealSource.initialize();
	}

	public void setCustomConfigurator(ConfigInfoIF config) {
		dealSource.setCustomConfigurator(config);
	}

	public boolean handleEvent(QueueElementIF event) throws Exception {
		if (event instanceof HeartbeatEvent) {
			// TODO
			// 
		} else if (event instanceof EntityNestedEvent) {

			EntityNestedEvent entityEvent = (EntityNestedEvent) event;
			EntityNestedEvent aEvent = null;

			// 这个有错误,不处理,直接专交下一个模块
			if (entityEvent.getErrorCode() != 0) {
				LOG.warn("命令对象有错误.详细信息是: " + entityEvent.getErrorMessage());
				LOG.warn(entityEvent.getNestedEntity());
				this.nextStage.putEvent(entityEvent);

				return true;
			}

			try {
				Object result = dealSource.deal(entityEvent.getNestedEntity());
				if (result != null) {
					aEvent = new EntityNestedEvent(result);
					this.nextStage.putEvent(aEvent);
				}
			} catch (Exception e) {
				entityEvent.setErrorCode(-1);
				entityEvent.setErrorMessage(e.getMessage());
				LOG.error("处理数据发生异常:", e);
				this.nextStage.putEvent(entityEvent);
			}
		} else if (event instanceof NullEvent) {

			LOG.trace("没有数据,线程暂停 " + sleepMilisecond + " 毫秒");
			try {

				Thread.currentThread().sleep(sleepMilisecond);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		} else if (event instanceof StageInterruptEvent) {
			// 要向后面的Stage转发InterruptEvent
			LOG.debug("接收到中断命令,将中断命令进行传递");
			this.nextStage.putEvent(new StageInterruptEvent());
			// Thread.currentThread().interrupt();
		} else {
			// 正常设计的话,应该不会出现这个错误
			throw new UnknownStageEventException("位置的命令类型["
					+ event.getClass().getName() + "].");
		}
		return true;

	}

	public DealSourceIF getDealSource() {
		return dealSource;
	}

	public void setDealSource(DealSourceIF dealSource) {
		this.dealSource = dealSource;
	}

	public StageWrapperIF getNextStage() {
		return nextStage;
	}

	public void setNextStage(StageWrapperIF nextStage) {
		this.nextStage = nextStage;
	}

	public long getSleepMilisecond() {
		return sleepMilisecond;
	}

	public void setSleepMilisecond(long sleepMilisecond) {

		this.sleepMilisecond = sleepMilisecond;
	}

}
